package com.itcj.dmp.report

import com.itcj.dmp.utils.KuduHelper
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, SparkSession}

trait CommonReport {

  //原表名
  def odsTablename:String
  //新表名
  def resultTableName:String


  //定义keys
  def  keys:List[String]

  //具体实现过程
  def processDf(sourceDf:DataFrame):DataFrame

  def schemaChange(schema: StructType):Unit

  def process()={
    //    1. 创建SparkSession
    import com.itcj.dmp.utils.SparkConfigHelper._
    val sparkSession: SparkSession =SparkSession.builder().master("local[2]").appName("RegionReportProcess")
      .loadConfig()
      .getOrCreate()

    //      2. 加载ods表的数据
    import com.itcj.dmp.utils.KuduHelper._
    val odf = sparkSession.readKuduTable(odsTablename)


    //      3. 按照省市分组,求得结果
    if (odf.isDefined) {
      val df: DataFrame = odf.get
      //      val numds = df.selectExpr("region","city","count() as num").groupBy("region","city")
      val redf: DataFrame = processDf(df)
      //    4. 结果数据保存到kudu表中
      val schema: StructType = redf.schema
      schemaChange(schema)

      sparkSession.createKuduTable(resultTableName, schema, keys)
      redf.saveKuduTable(resultTableName)
    }
  }
}
